package com.shujia.sql

import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.table.api.EnvironmentSettings
import org.apache.flink.table.api.bridge.scala.StreamTableEnvironment

object Demo03SQLOnKafka {
  def main(args: Array[String]): Unit = {
    val bsEnv: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    val bsSettings: EnvironmentSettings = EnvironmentSettings
      .newInstance()
      .useBlinkPlanner() // 使用Blink的SQL解析器
      .inStreamingMode() // 开启流处理模式
      .build()
    // 构建Table Env
    val bsTableEnv: StreamTableEnvironment = StreamTableEnvironment.create(bsEnv, bsSettings)

    bsEnv.setParallelism(1)

    /**
     * Mary,2022-06-18 11:02:00,/admin
     * Mary,2022-06-18 11:02:02,/login
     * Mary,2022-06-18 11:02:05,/logout
     * Mary,2022-06-18 11:32:05,/logout
     * Mary,2022-06-18 11:42:05,/logout
     * Mary,2022-06-18 11:52:05,/logout
     * Mary,2022-06-18 12:00:00,/logout
     * Mary,2022-06-18 12:00:05,/logout
     * Mary,2022-06-18 12:00:06,/logout
     * Mary,2022-06-18 12:00:06,/logout
     * Mary,2022-06-18 12:01:05,/logout
     * Mary,2022-06-18 12:02:06,/logout
     * Mary,2022-06-18 12:03:05,/logout
     * Mary,2022-06-18 12:50:06,/logout
     * Mary,2022-06-18 13:00:05,/logout
     */

    // 构建Source表 动态表
    bsTableEnv
      .executeSql(
        """
          |CREATE TABLE click_table (
          | `user` STRING,
          | cTime TIMESTAMP(3),
          | url STRING,
          | WATERMARK FOR cTime AS cTime - INTERVAL '5' SECOND -- 设置5s的水位线
          |) WITH (
          | 'connector' = 'kafka',
          | 'topic' = 'click1',
          | 'properties.bootstrap.servers' = 'master:9092,node1:9092,node2:9092',
          | 'properties.group.id' = 'test1',
          | 'format' = 'csv',
          | 'csv.field-delimiter' = ',',
          | 'csv.ignore-parse-errors' = 'true',
          | 'scan.startup.mode' = 'earliest-offset'
          |)
          |""".stripMargin)


    // 创建Sink表
    //    bsTableEnv
    //      .executeSql(
    //        """
    //          |CREATE TABLE print_table WITH ('connector' = 'print')
    //          |LIKE click_table (EXCLUDING ALL)
    //          |""".stripMargin)

    bsTableEnv
      .executeSql(
        """
          |create table click_cnt_print(
          |   `user` String,
          |   cnt BIGINT
          |) WITH (
          | 'connector' = 'print'
          |)
          |""".stripMargin)

    // 执行查询并将结果输出
    //    bsTableEnv.executeSql(
    //      """
    //        |insert into click_cnt_print
    //        |select user,count(url) as cnt from click_table group by user
    //        |""".stripMargin)

    // 每隔一个小时统计用户的点击次数

    bsTableEnv
      .executeSql(
        """
          |create table click_cnt_print_with_time_window(
          |   `user` String,
          |   cnt BIGINT,
          |   endT TIMESTAMP(3)
          |) WITH (
          | 'connector' = 'print'
          |)
          |""".stripMargin)

    bsTableEnv
      .executeSql(
        """
          |insert into click_cnt_print_with_time_window
          |select user
          |       ,count(url) as cnt
          |       ,TUMBLE_END(cTime, INTERVAL '1' HOURS) as endT
          |from click_table
          |group by user,TUMBLE(cTime, INTERVAL '1' HOURS)
          |""".stripMargin)


  }

}
